package org.example

import org.apache.commons.lang3.StringUtils
import org.apache.spark.internal.Logging

import java.sql.{Connection, DriverManager, PreparedStatement}

/**
 * @ClassName MySQLHelper
 * @Description MySQL数据存取
 * @Auther maolingwei
 * @Date 2020-08-24 16:18
 * @Version 1.0
 * */
object MySQLHelper extends Serializable with Logging{

  val driverClassName = "com.mysql.jdbc.Driver"
  val colums = Array("district","enterprise","originalMsg","vid","vin")
  //批量插入
   def insertBatch(array: List[Map[String, Any]], table: String, env: String): Unit = {

    Class.forName(driverClassName)

     println(array)
    val (user, password, url) = getConnectionConfig(env)
    var connection: Connection = null

    var prepareStatement: PreparedStatement = null
    //todo
    val sql = "INSERT IGNORE INTO tt (a, b, c, d, e) VALUES (?,?,?,?,?)"
    if(array.nonEmpty){

      try {
        connection = DriverManager.getConnection(url, user, password)

        connection.setAutoCommit(false)

        prepareStatement = connection.prepareStatement(sql)

        var rowCount = 0



        array.foreach(record => {
          (1 to colums.size).map(i => {
            println(record.getOrElse(colums(i-1),"tt"))
            prepareStatement.setObject(i, record.getOrElse(colums(i-1), "1"))
          })

          prepareStatement.addBatch()

          rowCount += 1
        //批量操作，最大数量1000
          if (rowCount % 1000 == 0) {
            prepareStatement.executeBatch()
            rowCount = 0
            logInfo("插入1000条数据...")
          }

        })
        if (rowCount != 0) {
          prepareStatement.executeBatch()
        }
        connection.commit()

      } catch {
        case e: Throwable =>
          logError(s"写入表${table}失败!", e)
      } finally {
        if (prepareStatement != null) {
          try {
            prepareStatement.close()

          } catch {
            case e: Throwable => ""

          }

        }

        connection.close()

      }
    }
  }
  //批量更新
  private def updateBatch(array: Array[Map[String, Any]], table: String, env: String): Unit = {

    Class.forName(driverClassName)
    val length = 5
    val (user, password, url) = getConnectionConfig(env)
    var connection: Connection = null

    var prepareStatement: PreparedStatement = null
    //todo
    val sql = "INSERT IGNORE INTO tt (a, b, c, d, e) VALUES (?,?,?,?,?)"
    if(array.nonEmpty){

      try {
        connection = DriverManager.getConnection(url, user, password)

        connection.setAutoCommit(false)

        prepareStatement = connection.prepareStatement(sql)

        var rowCount = 0

        val keys = array.head.keySet.toList

        val length = keys.length

        array.foreach(record => {
          (1 to length).map(i => {
            prepareStatement.setObject(i, record.getOrElse(keys(i - 1), null))
          })

          prepareStatement.addBatch()

          rowCount += 1
          //批量操作，最大数量1000
          if (rowCount % 1000 == 0) {
            prepareStatement.executeBatch()
            rowCount = 0
            logInfo("更新1000条数据...")
          }

        })
        if (rowCount != 0) {
          prepareStatement.executeBatch()
        }
        connection.commit()

      } catch {
        case e: Throwable =>
          logError(s"写入表${table}失败!", e)
      } finally {
        if (prepareStatement != null) {
          try {
            prepareStatement.close()

          } catch {
            case e: Throwable => ""

          }

        }

        connection.close()

      }
    }
  }



  def getConnectionConfig(env: String): (String, String, String) = {
    var user = "root"
    var password = "q123456"
    var url = "jdbc:mysql://10.41.22.77:3306/statistics?useUnicode=true&characterEncoding=utf-8&useSSL=false"

    (user, password, url)
  }
}
